package com.hngy.java.tablesql;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;

import java.util.ArrayList;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 将DataSet转换成表
 */
public class TableSetToTableJava {

    public static void main(String[] args) {
        //获取BatchTableEnvironment
        ExecutionEnvironment bEnv = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment bTableEnv = BatchTableEnvironment.create(bEnv);

        //获取DataSet
        ArrayList<Tuple2<Integer,String>> data = new ArrayList<>();
        data.add(new Tuple2<>(1,"jack"));
        data.add(new Tuple2<>(2,"tom"));
        data.add(new Tuple2<>(2,"mack"));
        DataSource<Tuple2<Integer, String>> set = bEnv.fromCollection(data);

        //第一种：将DataSet转换为view视图
        bTableEnv.createTemporaryView("myTable",set,$("id"),$("name"));
        bTableEnv.sqlQuery("select id,name from myTable where id > 1").execute().print();

        //第二种：将DataSet转换为table对象
        Table table = bTableEnv.fromDataSet(set, $("id"), $("name"));
        table.select($("id"),$("name"))
                .filter($("id").isGreater(1))
                .execute()
                .print();
    }
}
